package com.sh.rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.sh.rabbitmq.utils.ChannelFactory;

import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

public class Producer {

    //public static final String QUEUE_NAME = "ack";

    public static void main(String[] args) throws Exception {
        //单个确认
        //Producer.oneConfirm();

        //批量确认
        //Producer.batchConfirm();

        //异步确认
        Producer.batchConfirmAsnc();
    }

    static void oneConfirm() throws Exception {
        Channel channel = ChannelFactory.getChannel();


        String QUEUE_NAME = UUID.randomUUID().toString();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //开启发布确认
        channel.confirmSelect();

        int a = 1000;

        long begin = System.currentTimeMillis();
        for (int i = 0; i < a; i++) {
            String msg = i + "";
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            System.out.println("消息" + i);
            //单个确认
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息确认成功");
            }
        }
        long end = System.currentTimeMillis();

        System.out.println("单个确认耗时：" + (end - begin));
    }

    static void batchConfirm() throws Exception {
        Channel channel = ChannelFactory.getChannel();


        String QUEUE_NAME = UUID.randomUUID().toString();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //开启发布确认
        channel.confirmSelect();

        int a = 1000;

        long begin = System.currentTimeMillis();
        for (int i = 0; i < a; i++) {
            String msg = i + "";
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            System.out.println("消息" + i);
            //批量确认
            if (i % 100 == 0) {
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("消息确认成功");
                }
            }
        }
        long end = System.currentTimeMillis();

        System.out.println("单个确认耗时：" + (end - begin));
    }

    static void batchConfirmAsnc() throws Exception {
        Channel channel = ChannelFactory.getChannel();


        String QUEUE_NAME = UUID.randomUUID().toString();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //开启发布确认
        channel.confirmSelect();

        ConcurrentHashMap<String,Object> concurrentHashMap = new ConcurrentHashMap();

        ConfirmCallback confirmCallback = (var1, var3) -> {
            concurrentHashMap.remove(var1);
            System.out.println("消息已收到" + var1);
        };

        ConfirmCallback noConfirmCallback = (var1,var2) -> {
            System.out.println("未确认");
        };


        int a = 1000;

        long begin = System.currentTimeMillis();

        channel.addConfirmListener(confirmCallback,noConfirmCallback);
        for (int i = 0; i < a; i++) {
            String msg = i + "";
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());

            concurrentHashMap.put(String.valueOf(channel.getNextPublishSeqNo()),msg);
        }
        long end = System.currentTimeMillis();

        System.out.println("异步确认耗时：" + (end - begin));

        System.out.println(concurrentHashMap);
    }
}
